package com.shujia.dwd

import com.shujia.common.util.SparkTool
import org.apache.spark.sql._

object DWDResRegnMergeLocationMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._

    import org.apache.spark.sql.functions._


    val wcdr: DataFrame = spark
      .table("ods.ods_wcdr")
      .where($"day_id" === day_id)

    val oidd: DataFrame = spark
      .table("ods.ods_oidd")
      .where($"day_id" === day_id)

    val ddr: DataFrame = spark
      .table("ods.ods_ddr")
      .where($"day_id" === day_id)

    val dpi: DataFrame = spark
      .table("ods.ods_dpi")
      .where($"day_id" === day_id)


    /**
      * 提取OIDD、移动DPI和WCDR分钟粒度位置数据，对重复时间点数据进行二次清洗，
      * 完成后写入合并文件库；根据乒乓切换、超速数据等清洗原则对噪声数据进行清洗，
      * 完成后写入融合文件库；移动DDR数据在数据上传后延迟一天进行。
      * 将原始信令数据融合处理后，根据应用的需求和后期应用场景的需求，
      * 共保留12个字段。包括手机号、城市代码、业务开始时间、经纬度、BSID、对端号码、数据源和业务类型。
      *
      */

    val merge: Dataset[Row] = wcdr
      .union(oidd)
      .union(dpi)
      .union(ddr)

    /**
      * 需要对敏感字段脱敏
      *
      */

    val mergeLocation: DataFrame = merge.select(
      upper(md5(concat($"mdn", expr("'shujia'")))) as "mdn",
      $"start_time",
      $"county_id",
      $"longi",
      $"lati",
      $"bsid",
      $"grid_id",
      $"biz_type",
      $"event_type",
      $"data_source"
    )


    /**
      * 保存数据
      *
      */
    mergeLocation
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save("/daas/motl/dwd/dwd_res_regn_mergelocation_msk_d/day_id=" + day_id)


    //增加而非你去
    spark.sql(
      s"""
         |alter table dwd.dwd_res_regn_mergelocation_msk_d  add if not exists partition(day_id='$day_id')
         |
      """.stripMargin)

  }

}
